package common;

import org.apache.flink.avro.shaded.org.apache.avro.generic.GenericRecord;
import org.apache.flink.table.data.GenericMapData;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;

import java.util.HashMap;
import java.util.Map;

public class Utils {

    public static Object avroGet(GenericRecord avro, String field) {
        try {
            return avro.get(field);
        } catch (Exception e) {
            return null;
        }
    }

    public static RowData mkRow(Object record, String fieldsStr, RowData... others) {
        if (record == null) {
            return null;
        }
        if (!(record instanceof GenericRecord)) {
            return null;
        }
        GenericRecord gr = (GenericRecord) record;

        GenericRowData row = null ;
        int i = 0;
        if ( (fieldsStr != null) && (!fieldsStr.equals("")) ) {
            String[] fields = fieldsStr.split(",");
            row = new GenericRowData(fields.length + others.length);

            for (String field : fields) {
                if ("extensions".equals(field)) {
                    Map<?, ?> map = (Map<?, ?>) avroGet(gr, field);
                    Map<StringData, StringData> stringDataMap = new HashMap<>();
                    if (map != null) {
                        for (Object s : map.keySet()) {
                            stringDataMap.put(StringData.fromString(String.valueOf(s)), StringData.fromString(String.valueOf(map.get(s))));
                        }
                    }
                    row.setField(i++, new GenericMapData(stringDataMap));
                } else {
                    row.setField(i++, StringData.fromString(String.valueOf(avroGet(gr, field))));
                }
                // row.setField(i++, field);
            }
        } else {
            row = new GenericRowData(others.length);
        }
        for (RowData other : others) {
            row.setField(i++, other);
        }
        return row;
    }

    public static String exceptionToString(Throwable t) {
        StringBuilder message = new StringBuilder();
        message.append(t.toString());
        for (StackTraceElement element : t.getStackTrace()) {
            message.append("\n\tat ").append(element.toString());
        }
        message.append("\n");
        for (Throwable throwable : t.getSuppressed()) {
            message.append(exceptionToString(throwable));
        }
        message.append("\n");
        Throwable cause = t.getCause();
        if (cause != null) {
            message.append(exceptionToString(cause));
        }
        return message.toString();
    }
}

